package com.Redis.Thread;

import javax.naming.NamingException;

import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import com.Redis.Model.ReportData;
import com.Redis.Serviceimpl.ConsumerServicesImpl;
import com.Redis.Utils.JedisUtil;
import com.Redis.Utils.MessageUtil;
import com.Redis.Utils.RedisPool;
import com.Redis.Utils.SerialoizebleUtil;
import com.Redis.Utils.SerialoizebleUtilbak;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.exceptions.JedisConnectionException;

public class StartThread{
	private Logger logger = Logger.getLogger(this.getClass().getName());
	private static StartThread instance = null;
	private ThreadPoolTaskExecutor threadPoolTaskExecutor;
	public static byte[] rediskey = "key".getBytes();
	public static byte[] dstkey = "dstkey".getBytes();

	public StartThread(ThreadPoolTaskExecutor threadPoolTaskExecutor) {
		this.threadPoolTaskExecutor = threadPoolTaskExecutor;
	}

	public void startReport() {	      
	        while (true) {
	            try {
	            	byte[] bytes = null;
	            	logger.info("==========================="+JedisUtil.getSize(rediskey));
	                try {
	                	bytes = JedisUtil.brpoplpush(rediskey, dstkey, 0);	
					} catch (Exception e) {
						// TODO: handle exception
					}
	            	if(null == bytes) {
	                	continue;
	                }
	                try {
	                	MessageUtil msg = (MessageUtil) SerialoizebleUtil.unSerialize(bytes);
	                	if (msg != null) {	   
		                    ThreadPoolExecutor(msg.getReportData());
		                }else {
		                	continue;
		                }	
					} catch (Exception e) {
						// TODO: handle exception
					}	                	                
	                if(JedisUtil.getSize(rediskey)==0) {
	                	logger.info("=============任务分发完成============");
	                	break;
	                }
	                Thread.sleep(100);
	            } catch (InterruptedException e) {
	                e.printStackTrace();
	            }
	        }		
	}

	/**
	 * 多线程处理任务
	 */
	public void ThreadPoolExecutor(ReportData reportData) {		
		new Thread(new StartTaskThread(threadPoolTaskExecutor, reportData)).start();
	}

}
